package cn._51doit.live.udf;

import cn._51doit.live.pojo.ItemEventCount;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class HotGoodTopNWindowFunction extends ProcessWindowFunction<Integer, ItemEventCount, Tuple3<String, String, String>, TimeWindow>  {


    @Override
    public void process(Tuple3<String, String, String> key, Context context, Iterable<Integer> elements, Collector<ItemEventCount> out) throws Exception {

        String categoryId = key.f0;
        String productId = key.f1;
        String eventId = key.f2;
        Integer count = elements.iterator().next();
        long start = context.window().getStart();
        long end = context.window().getEnd();

        out.collect(new ItemEventCount(categoryId, productId, eventId, count, start, end));

    }
}
